package cn.doitedu.dao;

import cn.doitedu.pojo.LogBean;
import org.apache.directory.api.util.GeneralizedTime;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.Map;

public class UserProfileHbaseDaoImpl implements UserProfileDao {

    private Connection connection;

    @Override
    public void init(ParameterTool parameterTool) throws Exception{

        //初始化连接
        //1.创建Hbase的连接
        org.apache.hadoop.conf.Configuration hbaseConf = HBaseConfiguration.create();
        hbaseConf.set("hbase.zookeeper.quorum", parameterTool.getRequired("hbase.zookeeper.quorum"));
        hbaseConf.setInt("hbase.zookeeper.property.clientPort", parameterTool.getInt("hbase.zookeeper.property.clientPort", 2181));
        connection = ConnectionFactory.createConnection(hbaseConf);

    }

    @Override
    public Result queryUserProfile(LogBean bean, Map<String, String> userProfiles) throws Exception{
        Table table = null;
        Result result = null;
        try {
            table = connection.getTable(TableName.valueOf("eagle_profile"));
            Get get = new Get(Bytes.toBytes(bean.getDeviceId()));
            //添加过滤条件
            for (String key : userProfiles.keySet()) {
                get.addColumn(Bytes.toBytes("f"), Bytes.toBytes(key));
            }
            result = table.get(get);
        } finally {
            if (table != null)
                table.close();
        }
        return result;
    }

    @Override
    public void close() throws Exception{
        connection.close();
    }
}
